package com.chukun.flink.stream.connector.sink;

import com.chukun.flink.stream.bean.KafkaMess;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011;

import java.util.Properties;

/**
 * @author chukun
 * @version 1.0.0
 * @description kakfa sink操作
 * @createTime 2022年05月28日 18:42:00
 */
public class KafkaSink {

    /**
     * 创建kafka sink传输器
     * @return
     */
    public static SinkFunction<String> createKafkaStringSink() {
        Properties proProperties = new Properties();
        proProperties.put("bootstrap.servers", "linux01:9092");
        FlinkKafkaProducer011<String> kafkaProducer = new FlinkKafkaProducer011<>(
                "kafka_flink", new SimpleStringSchema(), proProperties);
        return kafkaProducer;
    }

    /**
     * 创建kafka sink传输器，自定义序列化工具
     * @return
     */
    public static SinkFunction<KafkaMess> createModeledKafkaSink() {
        Properties proProperties = new Properties();
        proProperties.put("bootstrap.servers", "linux01:9092");
        FlinkKafkaProducer011<KafkaMess> kafkaProducer = new FlinkKafkaProducer011<>(
                "kafka_flink", new KafkaSerializationSchema(), proProperties);
        return kafkaProducer;
    }
}
